import org.apache.spark.sql.{SparkSession, DataFrame}
// 导入隐式转换
import spark.implicits._

object SparkSQLDelete {
  def main(args: Array[String]): Unit = {
    // 创建SparkSession
    val spark = SparkSession.builder()
      .appName("Spark SQL Delete Example")
      .config("spark.master", "local")
      .getOrCreate()

    // 连接MySQL数据库
    val url = "jdbc:mysql://localhost:3306/sparksql_practice"
    val properties = new java.util.Properties()
    properties.setProperty("user", "lsz")
    properties.setProperty("password", "lsz")

    // 读取students表数据
    var studentsDF: DataFrame = spark.read.jdbc(url, "students", properties)

    // 删除所有年龄超过30岁的学生记录
    studentsDF = studentsDF.filter($"age" <= 30)

    // 将删除后的数据重新写入数据库
    studentsDF.write.mode("overwrite").jdbc(url, "students", properties)

    // 验证删除结果
    val deletedStudentsDF: DataFrame = spark.read.jdbc(url, "students", properties)
    println("删除后的学生信息：")
    deletedStudentsDF.show()

    // 删除特定ID的学生记录，并验证删除成功
    val studentIDToDelete = 1 // 假设要删除ID为1的学生记录
    studentsDF = studentsDF.filter($"student_id" =!= studentIDToDelete)
    studentsDF.write.mode("overwrite").jdbc(url, "students", properties)

    // 验证删除结果
    val finalStudentsDF: DataFrame = spark.read.jdbc(url, "students", properties)
    println(s"删除ID为 $studentIDToDelete 的学生记录后的学生信息：")
    finalStudentsDF.show()

    // 关闭SparkSession
    spark.stop()
  }
}

